package sql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**

 滚动窗口
 keyby 统计

 窗口的起始点是根据整点来算的，跟时间语义没什么关系

 */
public class D9_1KafakSource_Tumble {


    public static void main(String[] args) {


        Configuration flinkConf = new Configuration();
        flinkConf.setString("rest.port","9091");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(flinkConf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSql = "CREATE TABLE ods_tb ( " +
            " stime STRING," +
            " name STRING," +
            " val BIGINT," +
            " rowtime AS to_timestamp(stime)," +
            " WATERMARK FOR rowtime AS rowtime - interval '10' second" +
            ") WITH ( " +
            "  'connector' = 'kafka'," +
            "  'topic' = 'test'," +
            "  'properties.bootstrap.servers' = 'wsl:9092'," +
            "  'properties.group.id' = 'testGroup'," +
            "  'scan.startup.mode' = 'latest-offset'," +
            "  'format' = 'json'" +
            ")";

        String print = "CREATE TABLE print (" +"    " +
            "    window_start timestamp, " +
            "    window_end timestamp, " +
            "    name STRING, " +
            "    val BIGINT " +
            ") WITH (" +
            "     'connector' = 'print'" +
            ")";

        String sql = "insert into print " +
                " SELECT " +
            " window_start, window_end," +
            " name," +
            " sum(val) " +
            "FROM " +
            " table(tumble(table ods_tb, descriptor(rowtime), interval '1' minutes))" +
            " GROUP BY name" +
            ",window_start,window_end,window_time";


        tableEnv.executeSql(genSql);
        tableEnv.executeSql(print);

        System.out.println(sql);
        tableEnv.executeSql(sql);


        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
